package com.shujia.dws

import com.shujia.utils.SparkTool
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql._

object DwsCityTouristMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    //1、读取融合表
    val mergeLocation: Dataset[Row] = spark
      .table("dwi.dwi_res_regn_mergelocation_msk_d")
      .where($"day_id" === day_id)


    //2|读取用户画像表
    val usertag: Dataset[Row] = spark
      .table("dim.dim_usertag_msk_d")
      .where($"day_id" === day_id)


    //行政区配置表
    val adminCode: DataFrame = spark
      .table("dim.dim_admincode")


    /**
      * 游客的判断条件
      * 1、停留时间大于3小时
      * 2、出游距离大于10KM
      *
      */


    val resultDF: Dataset[Row] = mergeLocation
      //1、关联位置融合表和行政区配置表获取停留点的城市编号
      .join(adminCode.hint("broadcast"), "county_id")
      //计算用户在一个城市内的停留时间
      //获取一个人在一个城市内的第一个时间和最后一个时间
      .withColumn("min_date", min($"start_date") over Window.partitionBy($"mdn", $"city_id"))
      .withColumn("max_date", max($"end_date") over Window.partitionBy($"mdn", $"city_id"))
      .withColumn("d_stay_time", unix_timestamp($"max_date", "yyyyMMddHHmmss") - unix_timestamp($"min_date", "yyyyMMddHHmmss"))
      //1、停留时间大于3小时
      .where($"d_stay_time" > 60 * 60 * 3)

      //关联获取用户常住地的网格
      .join(usertag, "mdn")
      //计算用户常住地到停留点的距离
      .withColumn("distance", calculateLengthByGrid($"resi_grid_id", $"grid_id"))
      //计算最远距离
      .withColumn("d_max_distance", max($"distance") over Window.partitionBy($"mdn", $"city_id"))

      //2、出游距离大于10KM
      .where($"d_max_distance" > 10000)
      .select($"mdn", $"resi_county_id", $"city_id", round($"d_stay_time" / 3600, 2), round($"d_max_distance" / 1000, 2))
      .distinct()


    resultDF.write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_city_tourist_msk_d/day_id=$day_id")


    spark.sql(
      s"""
         |alter table dws.dws_city_tourist_msk_d add if not exists partition(day_id='$day_id')
        |
      """.stripMargin)
  }
}
